/*
 *	dfmultithreaded.c
 *	© 2009 DigiFi Limited
 *
 *	The core of multithreaded libdigifi and dfclient
 *
 */

#include "../headers/dfmultithread.h"

#include "util/workmanager.h"
#include "util/stringext.h"



enum dfmultithread_running {
	DFMULTITHREAD_STARTED,
	DFMULTITHREAD_RUNNING,
	DFMULTITHREAD_WAITING,
	DFMULTITHREAD_PROCESSING,
	DFMULTITHREAD_STOPPED,
	DFMULTITHREAD_ERROR
};

struct dfmultithread_threadstatus {
	pthread_t uid;
	unsigned int jobs_processed;
	enum dfmultithread_running status;
	char *buffer;
	unsigned int buffer_len;
};


// local functions
static void* dfmultithread_threadstart(void *arg);
static int dfmultithread_findthreadid(pthread_t uid);
static int dfmultithread_addtoqueue(unsigned int type, char *regex_name,
 				void (*callback)(void), char *command, unsigned int is_special, df_rows *rows);

static int dfmultithread_receive(int id, char *regex, char **receive, int group, int thread_id);
static int dfmultithread_send(int id, char *command);

static void dfmultithread_clearthreadbuffer(int thread_id);

static void dfmultithread_updatestatus(int thread_id, enum dfmultithread_running new_status);


// list of threads currently running
static struct dfmultithread_threadstatus *threads;
static int thread_count;

// bit array indicating whether threads are running
static unsigned int thread_running;

// bit to tell threads to shutdown
static unsigned int shutdown_bit;

// thread list mutex
pthread_mutex_t tstatus_mutex = PTHREAD_MUTEX_INITIALIZER;




// this function will setup a given number of threads to process
// work sent by the client
int dfmultithread_init(char *ipaddress, int num_threads) {
	
	int i, err;
	
	// fail if bigger than like 32?
	if (num_threads>(sizeof(unsigned int) * 8)) return -1;
	
	// setup the status bits
	thread_running = 0;
	shutdown_bit =0;
	
	thread_count = num_threads;
	
	// setup the work manager
	if ((err = wm_init_manager()) < 0) {return err;}
	
	// malloc us some space for status
	if ((threads = malloc(sizeof(struct dfmultithread_threadstatus) * num_threads)) == NULL) {return -DFCLIENT_ENOMEM;}
	
	
	// start up the threads
	pthread_mutex_lock(&tstatus_mutex);
	
	for (i=0; i<num_threads; i++) {
		threads[i].jobs_processed = 0;
		threads[i].status = DFMULTITHREAD_STARTED;
		threads[i].buffer = malloc(sizeof(char));
		threads[i].buffer[0] = '\0';
		threads[i].buffer_len = 0;
		
		err = pthread_create(&threads[i].uid, NULL, dfmultithread_threadstart, (void*)ipaddress);
		
		#ifdef DF_DEBUG
		dferrors_printdebug("Thread %lu started (%d).\n", threads[i].uid, err);
		#endif
		
		// check for an error
		if (err!=0) {
			// unlock status
			pthread_mutex_unlock(&tstatus_mutex);
			
			// shutdown threads
			dfmultithread_shutdown();
			return -1;
		}
	}
	
	wm_register_special(threads[0].uid);
	
	pthread_mutex_unlock(&tstatus_mutex);
		
	return 0;
}


int dfmultithread_connected(void) {
	// check that at least some of the threads are running
	// and that the rget + rows thread is running
	return (thread_running!=0) && ((thread_running & 1) == 1); 
}


int dfmultithread_shutdown(void) {
	
	dferrors_printdebug("Shutting down messaging threads\n");
	
	// set the shutdown bit
	shutdown_bit=1;
	
	// pulse the more jobs cond to wake up sleeping threads
	while (thread_running!=0) {dferrors_printdebug("."); pthread_cond_broadcast(&wm_more_jobs); sleep(1);}
	
	dferrors_printdebug(" Done.\n");
	
	return 0;
}










int dfmultithread_queue(unsigned int type, char *regex_name, void (*callback)(void), char *command) {
	return dfmultithread_addtoqueue(type, regex_name, callback, command, 0,NULL);
}

int dfmultithread_queue_rget(char *command, ...) {
	char *buffer;
	va_list arg_ptr;
	
	va_start(arg_ptr, command);
	vasprintf(&buffer, command, arg_ptr);
	va_end(arg_ptr);
	
	return dfmultithread_addtoqueue(DF_VOID_COMMAND, "void", NULL, buffer, 1, NULL);
}

int dfmultithread_queue_row(char *regex_name, void (*callback)(void), int id, int total, int count, int offset, char *command, ...) {
	char *buffer;
	va_list arg_ptr;
	df_rows *pull_info;
	
	va_start(arg_ptr, command);
	vasprintf(&buffer, command, arg_ptr);
	va_end(arg_ptr);
	
	pull_info = malloc(sizeof(df_rows));
	pull_info->id = id;
	pull_info->total = total;
	pull_info->count = count;
	pull_info->offset = offset;
	
	return dfmultithread_addtoqueue(DF_ROW_COMMAND, regex_name, callback, buffer, 1, pull_info);
}







void dfmultithread_dump(void) {
	int i;
	char *status;
	
	// dump info about each thread
	dferrors_printdebug("Thread list\n-----------------\n");
	for (i=0; i<thread_count; i++) {		
		// niceify the status int
		if (threads[i].status == DFMULTITHREAD_STARTED) {
			status = "Started";
		}
		else if (threads[i].status == DFMULTITHREAD_RUNNING) {
			status = "Running";
		}
		else if (threads[i].status == DFMULTITHREAD_WAITING) {
			status = "Waiting";
		}
		else if (threads[i].status == DFMULTITHREAD_PROCESSING) {
			status = "Processing";
		}
		else if (threads[i].status == DFMULTITHREAD_STOPPED) {
			status = "Stopped";
		}
		else if (threads[i].status == DFMULTITHREAD_ERROR) {
			status = "An error has occured";
		}
		else {
			status = "Unknown status, possibly corrupt.";
		}
		
		dferrors_printdebug("[%d]\t%lu\t%u\t%s\n", i, (long unsigned int)threads[i].uid, threads[i].jobs_processed, status);
	}
	dferrors_printdebug("\n");
	// dump job list
	wm_dump();
}

void dfmultithread_dumpbuffer(int thread_id) {
	dferrors_printdebug("Thread %d Buffer\n--------------------------\n%s\n",thread_id,threads[thread_id].buffer);
}







static int dfmultithread_addtoqueue(unsigned int type,
 			char *regex_name, void (*callback)(void), char *command, unsigned int is_special, df_rows *rows) {
	
	df_asyncpayload *payload;
	
	if ((payload = malloc(sizeof(df_asyncpayload))) == NULL) {return -DFCLIENT_ENOMEM;}
	payload->command = command;
	payload->type = type;
	if ((payload->regex_name = strdup(regex_name)) == NULL) {return -DFCLIENT_ENOMEM;}
	payload->callback = callback;
	payload->pull_info = rows;
	
	return wm_add_item((void*)payload, is_special);
}



static void* dfmultithread_threadstart(void *ipaddress) {
	
	int thread_id, socket_id, err;
	df_asyncpayload *payload;
	dfrx_result **rx;
	char *in;
	unsigned int i, command_count;
	
	// find out which thread we are in thread status
	thread_id = dfmultithread_findthreadid(pthread_self());
	
	dfmultithread_updatestatus(thread_id, DFMULTITHREAD_RUNNING);
	
	// setup the socket connection here
	socket_id = dfsocket_connect((char*)ipaddress, DF_SOPHIA_PORT);
	
	if (socket_id<0) {
		dfmultithread_updatestatus(thread_id, DFMULTITHREAD_ERROR);
	}
	
	// poll until we receive some data on the port
	if (dfsocket_poll(socket_id, DF_DEFAULT_TIMEOUT, 1) == DFSOCKET_TIMEOUT) {		
		dfmultithread_updatestatus(thread_id, DFMULTITHREAD_ERROR);
	}
	
	thread_running = thread_running | (1<<thread_id);
	
	// initally clear buffer to get rid of the welcome message
	dfmultithread_clearthreadbuffer(thread_id);
	
	
	while ((threads[thread_id].status!=DFMULTITHREAD_ERROR) && (shutdown_bit==0)) {
		#ifdef DF_DEBUG
		dferrors_printdebug("Thread [%lu]: Connected and waiting on work.\n", (long unsigned int)threads[thread_id].uid);
		#endif
		
		dfmultithread_updatestatus(thread_id, DFMULTITHREAD_WAITING);
		
		// wait on work item
		while (!wm_item_avaliable() && (shutdown_bit==0)) { pthread_cond_wait(&wm_more_jobs, &wm_more_jobs_mutex); }
		
		if (shutdown_bit == 1) {pthread_mutex_unlock(&wm_more_jobs_mutex); break;}
		
		// do the work
		payload = (df_asyncpayload *)wm_get_item();
		
		if (payload==NULL) {continue;}
		
		pthread_mutex_unlock(&wm_more_jobs_mutex);
		
		dfmultithread_updatestatus(thread_id, DFMULTITHREAD_PROCESSING);
		
		#ifdef DF_DEBUG
		dferrors_printdebug("Thread [%lu]: To send %s.\n", (long unsigned int)threads[thread_id].uid, payload->command);
		#endif
		
		if (payload->pull_info!=NULL) { command_count = payload->pull_info->count; }
		else { command_count = 1; }
		
		// allocate enough space for all the incoming parsed strings
		rx = malloc(sizeof(dfrx_result*) * command_count);
		
		// send the command
		dfmultithread_send(socket_id, payload->command);
		
		#ifdef DF_DEBUG
		dferrors_printdebug("Thread [%lu]: Sent.\n", (long unsigned int)threads[thread_id].uid);
		dferrors_printdebug("Thread [%lu]: Receiving %d rows.\n", (long unsigned int)threads[thread_id].uid, command_count);
		#endif
		
		for (i=0; i<command_count; i++) {			
			if ((err=dfmultithread_receive(socket_id, payload->regex_name, &in, payload->type, thread_id)) <0) {break;}
			
			#ifdef DF_DEBUG
			dferrors_printdebug("Thread [%lu]: Received (%d of %d): \"%s\".\n", (long unsigned int)threads[thread_id].uid,i,command_count, in);
			#endif
			
			rx[i] = dfregex_matches(in, payload->regex_name, payload->type);
			
			free(in);
		}
		
		#ifdef DF_DEBUG
		dferrors_printdebug("Thread [%lu]: Finished receive.\n", (long unsigned int)threads[thread_id].uid);
		#endif
		
		// clear the buffer
		dfmultithread_clearthreadbuffer(thread_id);
		
		// if we succeeded in receiving the rows
		if (err==0) {
			#ifdef DF_DEBUG
			dferrors_printdebug("Thread [%lu]: Callback with: %s.\n", (long unsigned int)threads[thread_id].uid, (rx[0]!=NULL) ? rx[0]->value : "(undefined)");
			#endif
		
			if (payload->callback!=NULL) {
				if (payload->type == DF_GET_COMMAND) {
					dfget_callbackhandler(payload->regex_name, rx[0], payload->callback);
				}
				else if (payload->type == DF_ROW_COMMAND) {
					if ((strcmp(payload->regex_name, "rowrequest") == 0) || (strcmp(payload->regex_name, "rowrequestroom") == 0)) {
						#ifdef DF_DEBUG
						dferrors_printdebug("Thread [%lu]: Getting associated rows.\n", (long unsigned int)threads[thread_id].uid);
						#endif
						dfrow_pullrows(rx[0], payload);
					}
					else {
						dfrow_callbackhandler(payload->regex_name, rx, payload->callback, payload->pull_info);
					}
				}
			}
			else {
				dferrors_printdebug("Thread [%lu]: Unwanted return: %s.\n", (long unsigned int)threads[thread_id].uid, (rx[0]!=NULL) ? rx[0]->value : "(undefined)");
			}
		}
		
		
		
		free(payload->regex_name);
		if (rx!=NULL) {
			for (i=0; i<command_count; i++) {
				dfregex_freeresult(rx[i]);
			}
		}
		free(payload);
		
		threads[thread_id].jobs_processed++;
	}
	
	// shutdown socket
	dfsocket_disconnect(socket_id);
	
	#ifdef DF_DEBUG
	dferrors_printdebug("Thread [%lu]: Exiting.\n", (long unsigned int)threads[thread_id].uid);
	#endif
	
	thread_running = thread_running & ~(1<<thread_id);
	
	dfmultithread_updatestatus(thread_id, DFMULTITHREAD_STOPPED);
	
	return NULL;
}


static int dfmultithread_send(int id, char *command) {
	int err;
	
	// try send
	err = dfsocket_send(id, command, strlen(command) * sizeof(char));
	
	if (err<(strlen(command) * sizeof(char))) {
		dfsocket_disconnect(id);
		return err;
	}
	
	return 0;
}


static int dfmultithread_receive(int id, char *regex, char **receive, int group, int thread_id) {
	int received, total_received;
	dfrx_result *match;
	
	// sent succesfully now wait for some incoming data
	total_received = 0;
	
	while (dfregex_match(threads[thread_id].buffer,regex,group)==NULL && dfregex_match(threads[thread_id].buffer,"error",group)==NULL) {
		if ((threads[thread_id].buffer = 
					realloc(threads[thread_id].buffer, (threads[thread_id].buffer_len + 251) * sizeof(char))) == NULL) {return -DFCLIENT_ENOMEM;}
		
		dfsocket_poll(id, DF_DEFAULT_TIMEOUT,1);
		
		received = dfsocket_receive(id, threads[thread_id].buffer + threads[thread_id].buffer_len, 250);
		if (received < 0) return -1;
		
		threads[thread_id].buffer[threads[thread_id].buffer_len+received+1] = '\0';
		
		threads[thread_id].buffer_len += received;
			
	}
	
	if ((threads[thread_id].buffer = realloc(threads[thread_id].buffer, (threads[thread_id].buffer_len+1) * sizeof(char))) == NULL) {return -DFCLIENT_ENOMEM;}
	threads[thread_id].buffer[threads[thread_id].buffer_len] = '\0';
	
	// check if its an error first
	if ((match = dfregex_match(threads[thread_id].buffer,"error",group)) != NULL) {
		dferrors_printerror(-1, "%s", match->value);
		return -1;
	}
	
	// now chop out the command
	match = dfregex_match(threads[thread_id].buffer,regex,group);	
	if ((*receive = malloc((match->length+1) * sizeof(char))) == NULL) {return -DFCLIENT_ENOMEM;}
	
	// extract receive from the holder
	string_extract(&threads[thread_id].buffer, receive, match->start, match->length);
	threads[thread_id].buffer_len -= match->length;
	
	// clear up and then free the buffer
	free(match);
	
	return 0;
}


static void dfmultithread_clearthreadbuffer(int thread_id) {
	// wipe buffer
	threads[thread_id].buffer = memset(threads[thread_id].buffer,0,threads[thread_id].buffer_len);
	
	// free it
	free(threads[thread_id].buffer);
	
	// create a new buffer
	threads[thread_id].buffer = malloc(sizeof(char));
	threads[thread_id].buffer[0] = '\0';
	threads[thread_id].buffer_len = 0;
}

static void dfmultithread_updatestatus(int thread_id, enum dfmultithread_running new_status) {
	
	threads[thread_id].status = new_status;
	
}


static int dfmultithread_findthreadid(pthread_t uid) {
	
	int i;
	
	for (i=0; i<thread_count; i++) {
		if (pthread_equal(threads[i].uid, uid)) return i;
	}
	
	return -1;
}